#include "rpc_channel_client.h"

namespace xlcomm {
namespace net {

RpcChannelClient::RpcChannelClient(net::EventLoop* loop,
                                   const net::InetAddress& server_addr)
    : loop_(loop), client_(loop, server_addr, "RpcChannelClient") {
  codec_.set_rpc_message_cb(
      std::bind(&RpcChannelClient::OnRpcMessage, this, _1, _2));
  client_.set_message_callback(
      std::bind(&RpcMessageCodec::OnMessage, codec_, _1, _2, _3));
}

void RpcChannelClient::CallMethod(
    const google::protobuf::MethodDescriptor* method,
    google::protobuf::RpcController* controller,
    const google::protobuf::Message* request,
    google::protobuf::Message* response, google::protobuf::Closure* done) {
  LOG_DEBUG << "RpcChannelClient::CallMethod==== in";
  static uint64_t id = 0;

  RpcMessage rpcmsg;
  rpcmsg.set_type(RPC_TYPE_REQUEST);
  rpcmsg.set_id(++id);
  rpcmsg.set_service(method->service()->full_name());
  rpcmsg.set_method(method->name());
  std::string reqstr;
  request->SerializeToString(&reqstr);
  rpcmsg.set_request(reqstr);
  codec_.SendRpcMessage(client_.connection(), rpcmsg);

  LOG_DEBUG << "RpcChannelClient::CallMethod==== rpcmsg.method:"
            << method->name() << " Send ok";

  OutstandingCall out{response, done};
  outstandings_[id] = out;
}

void RpcChannelClient::OnRpcMessage(const net::TcpConnectionPtr& conn,
                                    const RpcMessage& rpcmsg) {
  LOG_DEBUG << "RpcChannelClient::OnRpcMessage==== in";
  if (rpcmsg.type() != RPC_TYPE_RESPONSE) return;

  Outstandings::iterator it = outstandings_.find(rpcmsg.id());
  if (it != outstandings_.end()) {
    LOG_DEBUG << "RpcChannelClient::OnRpcMessage==== find";
    OutstandingCall out = it->second;
    out.response->ParseFromString(rpcmsg.response());
    if (out.done) out.done->Run();
  }
}

}  // namespace net
}  // namespace xlcomm